package cn.com.tarotframework.server.message;

import cn.com.tarotframework.utils.response.TarotResponseResultBody;
import cn.com.tarotframework.utils.response.TarotResult;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.constraints.NotBlank;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.Collections;

/**
 * copyright (C), 2022, 塔罗牌基础架构
 *
 * @author xincan
 * @version 0.0.1-SNAPSHOT
 * @program tarot-demo
 * @date 2024/10/11 19:07
 * @description MessageController
 */
@Validated
@Tag(name = "消息中心管理", description = "消息中心管理")
@RestController
@RequestMapping("message")
@TarotResponseResultBody
public class MessageController {

    @Operation(summary = "发送消息", method = "GET", description = "发送消息")
    @GetMapping("send")
    public TarotResult<String> send(
            @Parameter(name = "endpoint", description = "Proxy地址", example = "172.27.109.1:32180")
            @NotBlank(message = "Proxy地址不能为空")
            @RequestParam(name = "endpoint") String endpoint,
            @Parameter(name = "topic", description = "topic", example = "TestTopic1")
            @NotBlank(message = "topic不能为空")
            @RequestParam(name = "topic") String topic,
            @Parameter(name = "msg", description = "消息", example = "测试rocketmq集群")
            @NotBlank(message = "消息不能为空")
            @RequestParam(name = "msg") String msg) {

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        try {
            // 初始化Producer时需要设置通信配置以及预绑定的Topic。
            Producer producer = provider.newProducerBuilder()
                    .setTopics(topic)
                    .setClientConfiguration(configuration)
                    .build();
            // 普通消息发送。
            Message message = provider.newMessageBuilder()
                    .setTopic(topic)
                    // 设置消息索引键，可根据关键字精确查找某条消息。
                    .setKeys("messageKey")
                    // 设置消息Tag，用于消费端根据指定Tag过滤消息。
                    .setTag("messageTag")
                    // 消息体。
                    .setBody(msg.getBytes())
                    .build();

            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            System.out.println("Send message successfully, messageId=" + sendReceipt.getMessageId());
            producer.close();
        } catch (ClientException | IOException err) {

            System.out.println(err);
            return TarotResult.error("发送失败");
        }

        return TarotResult.success("发送成功", msg);
    }


    @Operation(summary = "接收消息", method = "GET", description = "接收消息")
    @GetMapping("receive")
    public TarotResult<String> receive(
            @Parameter(name = "endpoint", description = "Proxy地址", example = "172.27.109.1:32181")
            @NotBlank(message = "Proxy地址不能为空")
            @RequestParam(name = "endpoint") String endpoint,
            @Parameter(name = "topic", description = "topic", example = "TestTopic1")
            @NotBlank(message = "topic不能为空")
            @RequestParam(name = "topic") String topic,
            @Parameter(name = "msg", description = "消息", example = "测试rocketmq集群")
            @NotBlank(message = "消息不能为空")
            @RequestParam(name = "msg") String msg
    ) {
        try {

            final ClientServiceProvider provider = ClientServiceProvider.loadService();
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoint).build();
            // 订阅消息的过滤规则，表示订阅所有Tag的消息。
            String tag = "*";
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
            PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    // 设置消费者分组。
//                .setConsumerGroup(consumerGroup)
                    // 设置预绑定的订阅关系。
                    .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                    // 设置消费监听器。
                    .setMessageListener(messageView -> {
                        // 处理消息并返回消费结果。
                        System.out.println("Consume message successfully, messageId=" + messageView.getMessageId());
                        return ConsumeResult.SUCCESS;
                    })
                    .build();
            Thread.sleep(Long.MAX_VALUE);
            // 如果不需要再使用 PushConsumer，可关闭该实例。
            pushConsumer.close();
        } catch (ClientException | InterruptedException | IOException err) {
            System.out.println(err);
            return TarotResult.error("接收失败");
        }
        return TarotResult.success("接收成功", msg);
    }

}
